﻿using Microsoft.Extensions.Options;

using RabbitMQ.Client;

using RabbitMQSample.WebApi.RabbitMQLibs.Options;

namespace RabbitMQSample.WebApi.RabbitMQLibs
{
    public class RabbitMQConnectionFactory : IRabbitMQConnectionFactory, IDisposable
    {
        private readonly IOptions<RabbitMqConnectionOptions> _connectionOptions;
        private readonly IWebHostEnvironment _webHostEnvironment;
        private readonly ILogger<RabbitMQConnectionFactory> _logger;
        private readonly IAsyncConnectionFactory _asyncConnectionFactory;
        private readonly List<string> _addresses;
        private readonly SemaphoreSlim _connectionLock;
        private readonly SemaphoreSlim _writeShareChannelLock;
        private readonly SemaphoreSlim _readShareChannelLock;

        private IConnection _connection;
        private IModel _writeShareChannel;
        private IModel _readShareChannel;

        public RabbitMQConnectionFactory(IOptions<RabbitMqConnectionOptions> connectionOptions, IWebHostEnvironment webHostEnvironment, ILogger<RabbitMQConnectionFactory> logger)
        {
            _connectionOptions = connectionOptions;
            _webHostEnvironment = webHostEnvironment;
            _logger = logger;
            _addresses = _connectionOptions.Value.Addresses.Split(',').ToList();
            _connectionLock = new SemaphoreSlim(1, 1);
            _writeShareChannelLock = new SemaphoreSlim(1, 1);
            _readShareChannelLock = new SemaphoreSlim(1, 1);

            _asyncConnectionFactory = new ConnectionFactory
            {
                VirtualHost = _connectionOptions.Value.Vhost,
                UserName = _connectionOptions.Value.Username,
                Password = _connectionOptions.Value.Password,
                AmqpUriSslProtocols = System.Security.Authentication.SslProtocols.None,

                DispatchConsumersAsync = true,
                ConsumerDispatchConcurrency = _connectionOptions.Value.ConsumerDispatchConcurrency,
                RequestedHeartbeat = TimeSpan.FromSeconds(10), // 心跳间隔
                HandshakeContinuationTimeout = TimeSpan.FromMinutes(5),
                ContinuationTimeout = TimeSpan.FromMinutes(5),
                AutomaticRecoveryEnabled = true, // 自动重连
                TopologyRecoveryEnabled = true, // 拓扑重连
                NetworkRecoveryInterval = TimeSpan.FromSeconds(10),
            };
        }

        public void Dispose()
        {
            _writeShareChannel?.Dispose();
            _readShareChannel?.Dispose();
            _connection?.Dispose();

            _connectionLock?.Dispose();
            _writeShareChannelLock?.Dispose();
            _readShareChannelLock?.Dispose();
        }

        /// <summary>
        /// 获取共享渠道
        /// </summary>
        public async Task<IModel> GetShareChannelAsync(ChannelTypeEnum channelType, CancellationToken cancellationToken = default)
        {
            switch (channelType)
            {
                case ChannelTypeEnum.Read:
                    {
                        if (_readShareChannel != null && _readShareChannel.IsOpen) return _readShareChannel;

                        try
                        {
                            await _readShareChannelLock.WaitAsync(cancellationToken);
                            if (_readShareChannel != null && _readShareChannel.IsOpen) return _readShareChannel;

                            _readShareChannel = await InnelGetChannelAsync(cancellationToken);

                            return _readShareChannel;
                        }
                        catch (Exception ex)
                        {
                            _logger.LogCritical(ex, "RabbitMQ 连接失败");
                            return null;
                        }
                        finally
                        {
                            _readShareChannelLock.Release();
                        }
                    }
                case ChannelTypeEnum.Write:
                    {
                        if (_writeShareChannel != null && _writeShareChannel.IsOpen) return _writeShareChannel;

                        try
                        {
                            await _writeShareChannelLock.WaitAsync(cancellationToken);
                            if (_writeShareChannel != null && _writeShareChannel.IsOpen) return _writeShareChannel;

                            _writeShareChannel = await InnelGetChannelAsync(cancellationToken);

                            return _writeShareChannel;
                        }
                        catch (Exception ex)
                        {
                            _logger.LogCritical(ex, "RabbitMQ 连接失败");
                            return null;
                        }
                        finally
                        {
                            _writeShareChannelLock.Release();
                        }
                    }
                default:
                    return null;
            }
        }

        /// <summary>
        /// 获取渠道
        /// </summary>
        public async Task<IModel> GetChannelAsync(CancellationToken cancellationToken = default)
        {
            return await InnelGetChannelAsync(cancellationToken);
        }

        private async Task<IModel> InnelGetChannelAsync(CancellationToken cancellationToken = default)
        {
            IConnection connection = await GetShareConnectionAsync(cancellationToken);
            IModel channel = connection.CreateModel();

            //var basicProperties = channel.CreateBasicProperties();
            //basicProperties.Persistent = true;
            //basicProperties.DeliveryMode = 2;

            return channel;
        }

        /// <summary>
        /// 获取共享连接
        /// </summary>
        private async Task<IConnection> GetShareConnectionAsync(CancellationToken cancellationToken = default)
        {
            if (_connection != null && _connection.IsOpen) return _connection;

            try
            {
                await _connectionLock.WaitAsync(cancellationToken);
                if (_connection != null && _connection.IsOpen) return _connection;

                IList<AmqpTcpEndpoint> endpoints = _addresses.Select(address => new AmqpTcpEndpoint(new Uri($"amqp://{address}"))).ToList();
                _connection = _asyncConnectionFactory.CreateConnection(endpoints, $"{_webHostEnvironment.ApplicationName}:{Environment.MachineName}:{Guid.NewGuid():N}");
            }
            catch (RabbitMQ.Client.Exceptions.BrokerUnreachableException)
            {
                //_logger.LogCritical(ex, "连接发生错误，稍后重试");
                throw;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "RabbitMQ 连接失败");
                return null;
            }
            finally
            {
                _connectionLock.Release();
            }

            return _connection;
        }

        /// <summary>
        /// 获取连接
        /// </summary>
        public Task<IConnection> GetConnectionAsync(CancellationToken cancellationToken = default)
        {
            return Task.Run(() =>
            {
                IList<AmqpTcpEndpoint> endpoints = _addresses.Select(address => new AmqpTcpEndpoint(new Uri($"amqp://{address}"))).ToList();
                var connection = _asyncConnectionFactory.CreateConnection(endpoints, $"{_webHostEnvironment.ApplicationName}:{Environment.MachineName}:{Guid.NewGuid():N}");
                return connection;
            });
        }
    }
}
